package com.margo.project.curator.queue;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.queue.DistributedQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;

/**
 * IMPORTANT - We recommend that you do NOT use ZooKeeper for Queues
 * 
 * An implementation of the Distributed Queue ZK recipe. 
 * Items put into the queue are guaranteed to be ordered (by means of ZK's PERSISTENTSEQUENTIAL node). 
 * If a single consumer takes items out of the queue, they will be ordered FIFO. 
 * If ordering is important, use a LeaderSelector to nominate a single consumer.
 * 
 * QueueBuilder
 * QueueConsumer
 * QueueSerializer
 * DistributedQueue
 *
 */
public class MargoDistributedQueue {
	public static void main(String[] args) {
		CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181",
				new ExponentialBackoffRetry(1000, 3));
		client.start();

		QueueConsumer<String> consumer = createQueueConsumer();
		QueueSerializer<String> serializer = createQueueSerializer();
		QueueBuilder<String> builder = QueueBuilder.builder(client, consumer, serializer, "/queue");

		DistributedQueue<String> queue = builder.buildQueue();
		try {
			queue.start();
			queue.put("margo");
			queue.put("18");
			
			Thread.sleep(40000);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			CloseableUtils.closeQuietly(queue);
			CloseableUtils.closeQuietly(client);
//            CloseableUtils.closeQuietly(server);
		}

	}

	private static QueueSerializer<String> createQueueSerializer() {
		return new QueueSerializer<String>() {

			public byte[] serialize(String item) {
				return item.getBytes();
			}

			public String deserialize(byte[] bytes) {
				return new String(bytes);
			}

		};
	}

	private static QueueConsumer<String> createQueueConsumer() {

		return new QueueConsumer<String>() {

			public void stateChanged(CuratorFramework client, ConnectionState newState) {
				System.out.println("connection new state: " + newState.name());
			}

			public void consumeMessage(String message) throws Exception {
				System.out.println("consume one message: " + message);
			}

		};
	}
}
